package com.hmdp.config;

import cn.hutool.json.JSONUtil;
import com.hmdp.entity.VoucherOrder;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * kafka的异常处理机制，当kafka消费者处理失败时，会调用该方法，将redis中的库存补偿、下单用户id删除
 * @author xyf
 */
@Component
@Slf4j
public class KafkaDefaultListenerErrorHandler implements KafkaListenerErrorHandler {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException e) {
        ConsumerRecord<String, String> record = (ConsumerRecord<String, String>)message.getPayload();
        VoucherOrder voucherOrder = JSONUtil.toBean(record.value(), VoucherOrder.class);
        log.error("kafka异常:{}",e.getMessage());
        stringRedisTemplate.opsForValue().increment("seckill:stock:" + voucherOrder.getVoucherId());
        stringRedisTemplate.opsForSet().remove("seckill:order:" + voucherOrder.getVoucherId(), voucherOrder.getUserId().toString());
        return null;
    }
}
